﻿//***********************************************************************
// Assembly         : IGolovkine.AoA.Core
// Author           : Ilia
// Created          : 06-11-2011
//
// Last Modified By : Ilia
// Last Modified On : 01-07-2010
// Description      : 
//
// Copyright        : (c) Microsoft. All rights reserved.
//***********************************************************************
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.IO;
using System.Threading.Tasks;
using System.Threading;

namespace IGolovkine.AoA
{
    /// <summary>
    /// 
    /// </summary>
    /// <remarks></remarks>
    public sealed class AsyncStream : IChannel<byte[]>, IObservable<byte[]>, IObserver<byte[]>, IDisposable
    {
        /// <summary>
        /// 
        /// </summary>
        private readonly Stream _stream;
        /// <summary>
        /// 
        /// </summary>
        private readonly int _bufferSize;
        /// <summary>
        /// 
        /// </summary>
        private IObserver<byte[]> _observer;
        /// <summary>
        /// 
        /// </summary>
        private volatile bool _cancelled;

        /// <summary>
        /// Initializes a new instance of the <see cref="AsyncStream"/> class.
        /// </summary>
        /// <param name="stream">The stream.</param>
        /// <param name="bufferSize">Size of the buffer.</param>
        /// <remarks></remarks>
        public AsyncStream(Stream stream, int bufferSize)
        {
            _stream = stream;
            _bufferSize = bufferSize;
        }

        /// <summary>
        /// Subscribes the specified observer.
        /// </summary>
        /// <param name="observer">The observer.</param>
        /// <returns></returns>
        /// <remarks></remarks>
        public IDisposable Subscribe(IObserver<byte[]> observer)
        {
            _observer = observer;
            Task task = Task.Factory.StartNew(() =>
            {
                byte[] buffer = new byte[_bufferSize];
                int lenght = 0;
                try
                {
                    while (_cancelled == false && (lenght = _stream.Read(buffer, 0, buffer.Length)) != 0)
                    {
                        byte[] copy = new byte[lenght];
                        Array.Copy(buffer, copy, lenght);
                        observer.OnNext(copy);
                    }
                }
                catch (Exception ex)
                {
                    observer.OnError(ex);
                }
                finally
                {
                    observer.OnCompleted();
                }
            });

            return Disposable.Create(() =>
            {
                this.Dispose();
                task.Wait();
            });
        }

        /// <summary>
        /// Gets the output.
        /// </summary>
        /// <remarks></remarks>
        public IObservable<byte[]> Output
        {
            get { return this; }
        }

        /// <summary>
        /// Gets the input.
        /// </summary>
        /// <remarks></remarks>
        public IObserver<byte[]> Input
        {
            get { return this; }
        }

        /// <summary>
        /// Ons the completed.
        /// </summary>
        /// <remarks></remarks>
        void IObserver<byte[]>.OnCompleted()
        {
            this.Dispose();
        }

        /// <summary>
        /// Ons the error.
        /// </summary>
        /// <param name="exception">The exception.</param>
        /// <remarks></remarks>
        void IObserver<byte[]>.OnError(Exception exception)
        {
            if (_observer != null)
            {
                _observer.OnError(exception);
            }
            else
            {
                throw exception;
            }
        }

        /// <summary>
        /// Ons the next.
        /// </summary>
        /// <param name="value">The value.</param>
        /// <remarks></remarks>
        void IObserver<byte[]>.OnNext(byte[] value)
        {
            try
            {
                _stream.Write(value, 0, value.Length);
                _stream.Flush();
            }
            catch 
            {
                this.Dispose();
            }
        }

        /// <summary>
        /// Performs application-defined tasks associated with freeing, releasing, or resetting unmanaged resources.
        /// </summary>
        /// <remarks></remarks>
        public void Dispose()
        {
            _cancelled = true;
            _observer = null;
            _stream.Dispose();

        }
    }
}
